package mrdemo007;

import java.io.IOException;
import java.util.Arrays;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class TopNMap  extends Mapper<LongWritable, Text,IntWritable,IntWritable>{
	
	int len;
	int[] topN;

	@Override
	protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, IntWritable,IntWritable>.Context context)
			throws IOException, InterruptedException {
		/**
		 * 在map阶段，就进行topN的处理的，只发出topN的数据
		 * 到了reduce的阶段，同样的处理topN的处理
		 */
		String line = value.toString();
		
		//输入的数据行，长度大于0则输出
		if(line!=null && line.trim().length()>0) {
			 String[] split = line.split(",");
			 if(split.length == 4) {
				 int x = Integer.valueOf(split[2]);
				 add(x);
			 }
 		}
	}
	
	private void add(int num) {
		//加入num到topN数组，并排序，确保数组元素值是升序，这样数组的元素第1到最后一个元素是我们需要的数据
		//数组的topN[0] 先放置新进来的元素，然后排序（升序）
		topN[0] = num;
		Arrays.sort(topN);
	}
	
//	Called once at the beginning of the task.
	@Override
	protected void setup(Mapper<LongWritable, Text, IntWritable, IntWritable>.Context context)
			throws IOException, InterruptedException {
		//拿到配置得到的参数N，如果没有默认值5
		len = context.getConfiguration().getInt("N", 5);
		//topN数组的初始化都是0
		topN = new int[len + 1];
		
//		比如求： top5  
//		        x
//		topN = [0,1,2,3,4,5 ];
//		把这个topN数组，给他排序（升序）
//		所有的数据处理完毕后，把topN的索引从1开始到最后的这5个数输出
	}
	
	@Override
	protected void cleanup(Mapper<LongWritable, Text, IntWritable, IntWritable>.Context context)
			throws IOException, InterruptedException {
		 for(int i = 1 ; i < len+1; i++) {
			 System.out.println(topN[i] + "---" + topN[i]);
			 context.write(new IntWritable(topN[i]), new IntWritable(topN[i]));
		 }
	}
	 
}
